package spark.work

//import breeze.linalg.{max, sum}
//import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
//import org.apache.spark.sql.functions._

/**
  * Created by liuwei on 2017/11/8.
  */
object CaseConversion {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("ClearingRepeat").setMaster("local[4]")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new SQLContext(sc)
    val spark = SparkSession.builder

    val ss = spark.getOrCreate();
    val df = ss.createDataFrame(Seq(
//      ("张三", "女", "数学", 60, 15, "张老师"),
      ("afffDD", "sssSSS", "数学", 88, 15, ""),
//      ("张三", "女", "数学", 88, 15, "liu老师"),
      ("", "女", "语文", 70, 12, "李老师"),
      ("张三", "女", "物理", 50, 22, "张老师"),
      ("王二", "男", "数学", 50, 16, "赵老师"),
      ("王二", "男", "语文", 60, 20, "孙老师"),
      ("王二", "男", "历史", 50, 22, "李老师")
    )).toDF("name", "sex", "科目", "成绩", "排名", "监考老师")
    val titleNames = Seq("name", "sex", "科目", "成绩", "排名", "监考老师")
//    val columnNames = Seq("name")


//    val columnTagIdx = columnNames.map(titleNames.indexOf(_))


    val columnNames =  ArrayBuffer("name","sex")
    val upperOrLowers = ArrayBuffer("lower","upper")

//   selectedJsonArray.foreach(x => {
//      val columnName = x.column
//      val upperOrLower = x.upperOrLower
//      columnNames += columnName
//      upperOrLowers += upperOrLower
//
//      //      val columnIndex = titleNames.indexOf(columnName)
//      //      val a =  inputRdd.map(row=>{
//      //        if(row.isNullAt(columnIndex)) "" else if(UPPER.equals(upperOrLower)) row.getString(columnIndex).toUpperCase
//      //        else row.getString(columnIndex).toLowerCase
//      //        row.
//      //       })
//    })
    println("CaseConversionJob param" + columnNames.toString())
    println("CaseConversionJob upperOrLowers param" + upperOrLowers.toString())
    val outputRDD = df.rdd.map(row => {
      val rowArr = row.toSeq.toArray
      columnNames.indices.foreach(index => {
        val idx = titleNames.indexOf(columnNames(index))
        val cell = if(rowArr(idx)== null) "" else rowArr(idx).toString
        val upperOeLower = upperOrLowers(index)
        val rowStr = upperOeLower match {
          case "upper" => cell.toUpperCase
          case _ => cell.toLowerCase
        }
        rowArr(idx) = rowStr
      })
      Row.fromSeq(rowArr.toSeq)
    })
    val outputDf = ss.createDataFrame(outputRDD, df.schema)
    outputDf.show()

    //    println(groupByCondition)
  }







}
